package db.prob.operators.project;

import java.io.IOException;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import db.prob.io.LineParser;
import db.prob.io.PTuple;

public class MapperProject extends Mapper<LongWritable, Text, Text, Text> {
	public static final String TABLE_NAME   = "table_name";
	public static final String COLUMNS      = "columns";
	public static final String RESULT_TABLE = "result_table";
	public static final String PASS = "pass";
	
	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		PTuple tuple = LineParser.createPTuple(value.toString(), "\\t");
		Configuration conf = context.getConfiguration();
		String tableName = conf.get(TABLE_NAME);
		
		Text newKey = new Text();
		Text newVal;
		
		if (tuple.getTableName().equals(tableName)) {
			String[] projAttrIndex = conf.get(COLUMNS).split(",");
			String[] projAttr = new String[projAttrIndex.length];
			for (int i = 0; i < projAttrIndex.length; i++) {
				// Creating the projected tuple's attributes.
				projAttr[i] = tuple.getAttr()[Integer.parseInt(projAttrIndex[i])];
			}
			PTuple newTuple = new PTuple(tuple.getTableName(), tuple.getProbability(), projAttr);
			newKey.set(StringUtils.join(projAttr,'\t'));
			newVal = new Text(newTuple.toString());
		} else {
			newKey.set(PASS);
			newVal = value;			
		}

		context.write(newKey,newVal);		
	}
	
	
	 
}
